package com.zyx.pulsardemo.functions.sdk;

import org.apache.pulsar.functions.api.Context;
import org.apache.pulsar.functions.api.Function;
import org.slf4j.Logger;

/**
 * @author Yaxi.Zhang
 * @since 2021/10/8 18:00
 * desc: 实现Pulsar的Function接口
 */
public class EchoSdkFunction implements Function<String, String> {
    /**
     * 该接口定义了一个带有两个参数的 方法, 包括一个Context对象
     */
    @Override
    public String process(String input, Context ctx) throws Exception {
        Logger logger = ctx.getLogger();

        // 使用Context对象来获取输入主题的列表
        String inputTopics = String.join(", ", ctx.getInputTopics());

        // 使用Context对象来获取函数名
        String functionName = ctx.getFunctionName();

        String logMessage =
                String.format("A message with a value of \"%s\" has arrived "
                                + "on one of the following topics: %s\n",
                        input, inputTopics);
        // 成一条日志消息
        logger.info(logMessage);
        String metricName =
                String.format("function-%s-messages-received", functionName);

        // 记录用户定义的指标
        ctx.recordMetric(metricName, 1);
        return input;
    }
}
